package main

import (
	"encoding/json"
	"fmt"

	"github.com/Shopify/sarama"
)

func main() {

	config := sarama.NewConfig()
	config.Producer.RequiredAcks = sarama.WaitForAll
	config.Producer.Partitioner = sarama.NewRandomPartitioner
	config.Producer.Return.Successes = true
	config.Producer.Return.Errors = true
	config.Version = sarama.V0_11_0_2

	configc := sarama.NewConfig()
	configc.Consumer.Return.Errors = true
	configc.Version = sarama.V0_11_0_2

	// consumer
	consumer, _ := sarama.NewConsumer([]string{"10.10.9.20:9092"}, configc)

	defer consumer.Close()

	partition_consumer, _ := consumer.ConsumePartition("block", 0, sarama.OffsetOldest) //"block", 0, sarama.OffsetNewest

	defer partition_consumer.Close()

	fmt.Println("start")
	for {
		select {
		case msg := <-partition_consumer.Messages():
			b := Block2{}
			err := json.Unmarshal(msg.Value, &b)
			if err == nil {
				fmt.Println(msg.Offset, b)
			} else {
				fmt.Println(err)
			}

		case err := <-partition_consumer.Errors():
			fmt.Printf("err :%s\n", err.Error())
		}
	}

}

type Block2 struct {
	Count uint16   `json:"count"`
	IDs   []string `json:"ids"`
}
